// Copyright 2013 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package dfa.axis.v1_19.matchtables;

import com.google.api.ads.dfa.axis.factory.DfaServices;
import com.google.api.ads.dfa.axis.v1_19.Advertiser;
import com.google.api.ads.dfa.axis.v1_19.AdvertiserGroup;
import com.google.api.ads.dfa.axis.v1_19.AdvertiserGroupRecordSet;
import com.google.api.ads.dfa.axis.v1_19.AdvertiserGroupRemote;
import com.google.api.ads.dfa.axis.v1_19.AdvertiserGroupSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.AdvertiserRecordSet;
import com.google.api.ads.dfa.axis.v1_19.AdvertiserRemote;
import com.google.api.ads.dfa.axis.v1_19.AdvertiserSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.ApiException;
import com.google.api.ads.dfa.axis.v1_19.CustomSpotlightVariableConfiguration;
import com.google.api.ads.dfa.axis.v1_19.SpotlightActivity;
import com.google.api.ads.dfa.axis.v1_19.SpotlightActivityGroup;
import com.google.api.ads.dfa.axis.v1_19.SpotlightActivityGroupRecordSet;
import com.google.api.ads.dfa.axis.v1_19.SpotlightActivityGroupSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.SpotlightActivityRecordSet;
import com.google.api.ads.dfa.axis.v1_19.SpotlightActivitySearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.SpotlightConfiguration;
import com.google.api.ads.dfa.axis.v1_19.SpotlightRemote;
import com.google.api.ads.dfa.lib.client.DfaSession;

import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;

/**
 * Implementation of LookupObject that will generate Advertiser match tables. Extends
 * AbstractSimpleLookupObject to handle aggregation of Advertiser IDs. Will generate 
 * spotlight activity data for all advertiser IDs in DT file.
 * 
 * @author Moshe Kahan
 */
class AdvertiserLookup extends AbstractSimpleLookupObject {
  private ThreadLocal<AdvertiserRemote> advertiserService;
  private ThreadLocal<AdvertiserGroupRemote> advertiserGroupService;
  private ThreadLocal<SpotlightRemote> spotlightService;
  private TableSettings tableSettings;

  // Progress indicator
  private int spotLightActivityCount;

  /**
   * Constructor accepts DFA service/session objects needed to create SOAP remotes to the
   * API.
   *
   * @param dfaServices DFA services object needed to create SOAP remotes to the API
   * @param session DFA session object needed to create SOAP remotes to the API
   * @param tableSettings match table application configuration used as a MatchTable object factory
   */
  AdvertiserLookup(DfaServices dfaServices, DfaSession session, TableSettings tableSettings) {
    // Request the advertiser service from the service client factory.
    super(LookupField.AdvertiserId.getFieldName());
    advertiserService = initializeService(dfaServices, session, AdvertiserRemote.class);
    advertiserGroupService = initializeService(dfaServices, session, AdvertiserGroupRemote.class);
    spotlightService = initializeService(dfaServices, session, SpotlightRemote.class);
    this.tableSettings = tableSettings;
    spotLightActivityCount = 0;
  }

  private <T> ThreadLocal<T> initializeService(final DfaServices dfaServices,
      final DfaSession session, final Class<T> interfaceClass) {
    return new ThreadLocal<T>() {
      protected T initialValue() {
        return dfaServices.get(session, interfaceClass);
      }
    };
  }

  /**
   * Called when parsing/aggregation is complete. Will lookup via SOAP all unique DFA entries
   * in the DT file and pass data to MatchTable class.
   *
   * @param outputPath location of match table to be generated.
   */
  public void generateMatchTable(String outputPath)
      throws MatchTableFileException, DfaRemoteException {
    List<MatchTable> matchTableList = new ArrayList<MatchTable>();
    MatchTable advertiserMatchTable =
        createMatchTable(MatchTable.Type.ADVERTISER, matchTableList);
    MatchTable spotlightCatMatchTable =
        createMatchTable(MatchTable.Type.ACTIVITY_CAT, matchTableList);
    MatchTable spotlightTypeMatchTable =
        createMatchTable(MatchTable.Type.ACTIVITY_TYPE, matchTableList);
    MatchTable userDefinedSpotVarsMatchTable =
        createMatchTable(MatchTable.Type.USER_DEFINED_SPOT_VARS, matchTableList);

    for (MatchTable matchTable : matchTableList) {
      matchTable.open(outputPath);
    }

    // We call generateAdvertiserMatchTable even if advertiserMatchTable is not needed to
    // generate set of spot IDs.
    Set<Long> spotIdSet = new HashSet<Long>();

    generateAdvertiserMatchTable(advertiserMatchTable, spotIdSet);

    if (spotlightCatMatchTable != null || spotlightTypeMatchTable != null ||
        userDefinedSpotVarsMatchTable != null) {
      generateSpotlightMatchTables(spotlightCatMatchTable, spotlightTypeMatchTable,
          userDefinedSpotVarsMatchTable, spotIdSet);
    }
  }

  // Utility method to both create a MatchTable and add it to a List if indeed the factory 
  // generated a non-null reference.  TableSettings will return null for match table types
  // that are not configured to be created.
  private MatchTable createMatchTable(MatchTable.Type type, List<MatchTable> matchTableList) {
    MatchTable table = tableSettings.createMatchTable(type);
    if (table != null) {
      matchTableList.add(table);
    }
    return table;
  }

  // Method to generate advertiser match table specifically.
  private void generateAdvertiserMatchTable(MatchTable advertiserMatchTable,
      Set<Long> spotIdSet) throws MatchTableFileException, DfaRemoteException {
    List<Advertiser> advertisers = new ArrayList<Advertiser>();
    Map<Long, AdvertiserGroup> advertiserGroupMap = new HashMap<Long, AdvertiserGroup>();

    SoapPageableLookup<Advertiser, AdvertiserSearchCriteria> advertiserSoapLookup =
        new SoapPageableLookup<Advertiser, AdvertiserSearchCriteria>(entityIds,
            new AdvertiserSoapLookupCallback());

    // Retrieve the advertisers and store them and the spotlight IDs. Aggregate advertiser group
    // IDs for SOAP lookup.
    while (advertiserSoapLookup.hasNext()) {
      Advertiser result = advertiserSoapLookup.next();
      advertisers.add(result);
      if (result.getAdvertiserGroupId() > 0) {
        advertiserGroupMap.put(result.getAdvertiserGroupId(), null);
      }
      if (result.getSpotId() > 0) {
        spotIdSet.add(result.getSpotId());
      }
    }

    if (advertiserMatchTable != null) {
      advertiserMatchTable.printProgress();
      // Lookup all stored advertiser group IDs.
      SoapPageableLookup<AdvertiserGroup, AdvertiserGroupSearchCriteria> soapLookupAdvertiserGroup =
          new SoapPageableLookup<AdvertiserGroup, AdvertiserGroupSearchCriteria>(
              advertiserGroupMap.keySet(), new AdvertiserGroupSoapLookupCallback());

      while (soapLookupAdvertiserGroup.hasNext()) {
        AdvertiserGroup advertiserGroup = soapLookupAdvertiserGroup.next();
        advertiserGroupMap.put(advertiserGroup.getId(), advertiserGroup);
      }

      // Iterate over stored advertisers now that we have the group IDs.
      for (Advertiser result : advertisers) {
        AdvertiserGroup advertiserGroup =
          advertiserGroupMap.get(result.getAdvertiserGroupId());
        advertiserMatchTable.write(
            Long.toString(result.getSpotId()),
            Long.toString(result.getId()),
            result.getName(),
            Long.toString(result.getAdvertiserGroupId()),
            (advertiserGroup != null) ? advertiserGroup.getName() : "");
      }
      advertiserMatchTable.close();
    }
  }

  // Method to generate spotlight activity and user defined variables match tables.
  private void generateSpotlightMatchTables(MatchTable spotlightCatMatchTable,
      MatchTable spotlightTypeMatchTable,
      MatchTable userDefinedSpotVarsMatchTable,
      Set<Long> spotIdSet) throws MatchTableFileException, DfaRemoteException {

    MatchTable.printProgress(spotlightCatMatchTable, spotlightTypeMatchTable,
        userDefinedSpotVarsMatchTable);
    final Object spotLightActivityCountLock = new Object();
    spotLightActivityCount = 0;

    Map<Long, List<SpotlightActivityGroup>> spotIdToSpotlightActivityGroupResults =
        new HashMap<Long, List<SpotlightActivityGroup>>();
    List<SpotlightConfiguration> spotlightConfigurationResults =
        new ArrayList<SpotlightConfiguration>();
    Map<Long, List<SpotlightActivity>> spotIdTospotlightActivities =
        new HashMap<Long, List<SpotlightActivity>>();

    // Retrieving all the Spotlight info can take a while.  Using concurrent threads.
    List<Future<SpotlightConfiguration>> taskResults =
        new ArrayList<Future<SpotlightConfiguration>>();
    // Set up executor.
    ExecutorService executor =
        Executors.newFixedThreadPool(TableSettings.API_CLIENT_THREADS, new ThreadFactory() {
          public Thread newThread(Runnable runnable) {
            Thread thread = Executors.defaultThreadFactory().newThread(runnable);
            thread.setDaemon(true);
            return thread;
          }
        });
    // Add jobs.
    for (Long spotId : spotIdSet) {
      final Long spotIdFinal = spotId;
      final List<SpotlightActivity> spotlightActivityResults = new ArrayList<SpotlightActivity>();
      final List<SpotlightActivityGroup> spotlightActivityGroupResults =
          new ArrayList<SpotlightActivityGroup>();
      spotIdToSpotlightActivityGroupResults.put(spotId, spotlightActivityGroupResults);
      spotIdTospotlightActivities.put(spotId, spotlightActivityResults);
      taskResults.add(executor.submit(new Callable<SpotlightConfiguration>() {
        public SpotlightConfiguration call() throws DfaRemoteException {
          SpotlightConfiguration configuration = 
              retrieveSpotlightActivities(spotIdFinal, spotlightActivityResults,
              spotlightActivityGroupResults);
          synchronized (spotLightActivityCountLock) {
            ++spotLightActivityCount;
            System.out.printf("\r[%d spotlight activity SOAP lookups completed]",
                spotLightActivityCount);
          }
          return configuration;
        }
      }));
    }
    // Wait for jobs to complete.
    for (Future<SpotlightConfiguration> future : taskResults) {
      try {
        spotlightConfigurationResults.add(future.get());
      } catch (Exception e) {
        if (e.getCause() instanceof DfaRemoteException) {
          throw (DfaRemoteException) e.getCause();
        }
        throw new DfaRemoteException("Error in contacting DFA Remote API", e);
      }
    }

    // Generate activity "cat" and "type" match tables.
    for (Map.Entry<Long, List<SpotlightActivity>> entry : spotIdTospotlightActivities.entrySet()) {
      Map<Long, SpotlightActivityGroup> spotlightActivityGroupMap =
          new HashMap<Long, SpotlightActivityGroup>();
      List<SpotlightActivityGroup> spotlightActivityGroupResults =
          spotIdToSpotlightActivityGroupResults.get(entry.getKey());
      if (spotlightActivityGroupResults != null) {
        for (SpotlightActivityGroup spotlightActivityGroup : spotlightActivityGroupResults) {
          spotlightActivityGroupMap.put(spotlightActivityGroup.getId(), spotlightActivityGroup);
        }
        // Generate activity "type" or activity groups.
        if (spotlightTypeMatchTable != null) {
          for (SpotlightActivityGroup spotlightActivityGroup : spotlightActivityGroupResults) {
            spotlightTypeMatchTable.write(
                Long.toString(spotlightActivityGroup.getSpotlightConfigurationId()),
                Long.toString(spotlightActivityGroup.getId()),
                spotlightActivityGroup.getTagString(),
                spotlightActivityGroup.getName());
          }
        }
      }
      // Generate activity "cat" match tables.
      for (SpotlightActivity spotlightActivity : entry.getValue()) {
        if (spotlightCatMatchTable != null) {
          SpotlightActivityGroup spotlightActivityGroup =
              spotlightActivityGroupMap.get(spotlightActivity.getActivityGroupId());
          spotlightCatMatchTable.write(
              Long.toString(entry.getKey()),
              Long.toString(spotlightActivity.getActivityGroupId()),
              spotlightActivityGroup.getTagString(),
              Long.toString(spotlightActivity.getId()),
              spotlightActivity.getTagString(),
              spotlightActivity.getName(),
              Long.toString(spotlightActivity.getTagMethodTypeId()));
        }
      }
    }
    if (spotlightTypeMatchTable != null) {
      spotlightTypeMatchTable.flush();
    }
    if (spotlightCatMatchTable != null) {
      spotlightCatMatchTable.flush();
    }

    // Generate user defined spotlight variables from already retrieved spotlight configuration.
    if (userDefinedSpotVarsMatchTable != null) {
      for (SpotlightConfiguration spotlightConfiguration : spotlightConfigurationResults) {
        CustomSpotlightVariableConfiguration[] customSpotlightVariableConfigurationArray =
            spotlightConfiguration.getCustomSpotlightVariableConfigurations();
        if (customSpotlightVariableConfigurationArray != null) {
          for (CustomSpotlightVariableConfiguration customSpotlightVariableConfiguration :
            customSpotlightVariableConfigurationArray) {
            userDefinedSpotVarsMatchTable.write(
                Long.toString(spotlightConfiguration.getId()),
                "u" + Long.toString(customSpotlightVariableConfiguration.getId()),
                customSpotlightVariableConfiguration.getName());
          }
        }
      }
      userDefinedSpotVarsMatchTable.flush();
    }

    if (spotLightActivityCount > 0) {
      System.out.printf("\r[%d spotlight activity SOAP lookups completed]\n",
          spotLightActivityCount);
    }

    if (spotlightCatMatchTable != null) {
      spotlightCatMatchTable.close();
    }
    if (spotlightTypeMatchTable != null) {
      spotlightTypeMatchTable.close();
    }
    if (userDefinedSpotVarsMatchTable != null) {
      userDefinedSpotVarsMatchTable.close();
    }
  }

  private SpotlightConfiguration retrieveSpotlightActivities(Long spotId,
      List<SpotlightActivity> spotlightActivityResults,
      List<SpotlightActivityGroup> spotlightActivityGroupResults) throws DfaRemoteException {
    SpotlightSoapLookupCallback spotlightSoapLookupCallback =
        new SpotlightSoapLookupCallback(spotId);
    SoapPageableLookup<SpotlightActivity, SpotlightActivitySearchCriteria> soapLookup =
        new SoapPageableLookup<SpotlightActivity, SpotlightActivitySearchCriteria>(
            spotlightSoapLookupCallback);
    // Iterate through activity "cat".
    while (soapLookup.hasNext()) {
      SpotlightActivity result = soapLookup.next();
      spotlightActivityResults.add(result);
    }

    List<SpotlightActivityGroup> spotlightActivityGroups = new ArrayList<SpotlightActivityGroup>();
    SpotlightGroupSoapLookupCallback spotlightGroupSoapLookupCallback =
        new SpotlightGroupSoapLookupCallback(spotId);
    SoapLookup<SpotlightActivityGroup> spotlightActivityGroupSoapLookup =
        new SoapLookup<SpotlightActivityGroup>(spotlightGroupSoapLookupCallback);
    while (spotlightActivityGroupSoapLookup.hasNext()) {
      SpotlightActivityGroup result = spotlightActivityGroupSoapLookup.next();
      spotlightActivityGroupResults.add(result);
    }
    return spotlightGroupSoapLookupCallback.getSpotlightConfiguration();
  }


  // SOAP LookupCallbacks
  private class AdvertiserSoapLookupCallback
    implements SoapPageableLookupCallback<Advertiser, AdvertiserSearchCriteria> {

    public int getRecords(AdvertiserSearchCriteria criteria, List<Advertiser> recordList)
        throws ApiException, RemoteException {
      AdvertiserRecordSet records = advertiserService.get().getAdvertisers(criteria);
      int pages = records.getTotalNumberOfPages();
      for (Advertiser record : records.getRecords()) {
        recordList.add(record);
      }
      return pages;
    }

    public AdvertiserSearchCriteria getCriteria() {
      return new AdvertiserSearchCriteria();
    }
  }

  private class AdvertiserGroupSoapLookupCallback
    implements SoapPageableLookupCallback<AdvertiserGroup, AdvertiserGroupSearchCriteria> {

    public int getRecords(AdvertiserGroupSearchCriteria criteria, List<AdvertiserGroup> recordList)
        throws ApiException, RemoteException {
      AdvertiserGroupRecordSet records =
          advertiserGroupService.get().getAdvertiserGroups(criteria);
      int pages = records.getTotalNumberOfPages();
      for (AdvertiserGroup record : records.getRecords()) {
        recordList.add(record);
      }
      return pages;
    }

    public AdvertiserGroupSearchCriteria getCriteria() {
      return new AdvertiserGroupSearchCriteria();
    }
  }

  // SpotlightSoapLookupCallback keyed to one spotlight id.
  private class SpotlightSoapLookupCallback implements
    SoapPageableLookupCallback<SpotlightActivity, SpotlightActivitySearchCriteria> {

    private long spotId;

    SpotlightSoapLookupCallback(long spotId) {
      this.spotId = spotId;
    }

    public int getRecords(SpotlightActivitySearchCriteria criteria,
        List<SpotlightActivity> recordList) throws ApiException, RemoteException {
      SpotlightActivityRecordSet records = spotlightService.get().getSpotlightActivities(criteria);
      int pages = records.getTotalNumberOfPages();
      for (SpotlightActivity record : records.getRecords()) {
        recordList.add(record);
      }
      return pages;
    }

    public SpotlightActivitySearchCriteria getCriteria() {
      SpotlightActivitySearchCriteria spotlightActivitySearchCriteria =
          new SpotlightActivitySearchCriteria();
      spotlightActivitySearchCriteria.setAdvertiserId(spotId);
      return spotlightActivitySearchCriteria;
    }
  }

  // SpotlightGroupSoapLookupCallback keyed to one spotlight id to retrieve spotlight groups.
  // First request will also retrieve spotlight configuration.
  private class SpotlightGroupSoapLookupCallback
    implements SoapLookupCallback<SpotlightActivityGroup> {

    private long spotId;
    private SpotlightConfiguration spotlightConfiguration;

    SpotlightGroupSoapLookupCallback(long spotId) {
      this.spotId = spotId;
    }

    SpotlightConfiguration getSpotlightConfiguration() {
      return spotlightConfiguration;
    }

    public int getRecords(long[] ids, List<SpotlightActivityGroup> recordList)
        throws ApiException, RemoteException {
      spotlightConfiguration = spotlightService.get().getSpotlightConfiguration(spotId);
      SpotlightActivityGroupSearchCriteria spotlightActivityGroupSearchCriteria =
          new SpotlightActivityGroupSearchCriteria();
      spotlightActivityGroupSearchCriteria.setAdvertiserId(spotId);
      SpotlightActivityGroupRecordSet spotlightActivityGroupRecordSet =
          spotlightService.get().getSpotlightActivityGroups(spotlightActivityGroupSearchCriteria);
      int count = 0;
      for (SpotlightActivityGroup spotlightActivityGroup :
        spotlightActivityGroupRecordSet.getRecords()) {
        ++count;
        recordList.add(spotlightActivityGroup);
      }
      return count;
    }
  }
}
